package com.qingyunge.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.qingyunge.bean.FenLenskuNumBean;
import com.qingyunge.bean.SkuClassBean;
import com.qingyunge.util.MyClickHouseUtil;
import com.qingyunge.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.time.Duration;

public class Dws_keyby_skuclass_window {
    
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        String topic = "sku_eval_topic";
        String gourpid = "Dws_keyby_skuclass_window";
        DataStreamSource<String> skuEvalDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, gourpid));
        SingleOutputStreamOperator<SkuClassBean> skuClassDS = skuEvalDS.flatMap(new FlatMapFunction<String, SkuClassBean>() {
            @Override
            public void flatMap(String s, Collector<SkuClassBean> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    Long id = jsonObject.getLong("id");
                    String skuId = jsonObject.getString("skuId");
                    String skuClass = skuId.substring(0, 1);
                    String name = jsonObject.getString("name");
                    Double price = jsonObject.getDouble("price");
                    Long ts = jsonObject.getLong("ts");
                    SkuClassBean skuClassBean = new SkuClassBean(id, skuId, skuClass, name, price, ts,1L);
                    collector.collect(skuClassBean);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        SingleOutputStreamOperator<SkuClassBean> skuClassBeanDS = skuClassDS.assignTimestampsAndWatermarks(WatermarkStrategy.<SkuClassBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<SkuClassBean>() {
            @Override
            public long extractTimestamp(SkuClassBean skuClassBean, long l) {
                return skuClassBean.getTs();
            }
        }));
        // TODO 品牌分类的评论量
        SingleOutputStreamOperator<FenLenskuNumBean> fengleipinglunMapDS = skuClassBeanDS.keyBy(new KeySelector<SkuClassBean, Tuple2<String,String>>() {
            @Override
            public Tuple2<String, String> getKey(SkuClassBean skuClassBean) throws Exception {
                return Tuple2.of(skuClassBean.getSkuClass(), skuClassBean.getSkuId());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<SkuClassBean>() {
                    @Override
                    public SkuClassBean reduce(SkuClassBean skuClassBean, SkuClassBean t1) throws Exception {
                        skuClassBean.setNum(skuClassBean.getNum() + t1.getNum());
                        return skuClassBean;
                    }
                }).map(new RichMapFunction<SkuClassBean, FenLenskuNumBean>() {
                    @Override
                    public FenLenskuNumBean map(SkuClassBean skuClassBean) throws Exception {
                        return new FenLenskuNumBean(skuClassBean.getSkuClass(),skuClassBean.getSkuId(), skuClassBean.getNum(),skuClassBean.getTs());
                    }
                });
        fengleipinglunMapDS.print("fenlei===");
        // TODO 写入cllikhouse
        fengleipinglunMapDS.addSink(MyClickHouseUtil.getSinkFunction("INSERT INTO dws_sp_pl_sku_class_agg (sku_class,skuId, count,ts) VALUES (?,?,?,?)"));
        env.execute("Dws_keyby_skuclass_window");
    }
}